package com.gwtt.simulator.netconf.subsystem;

import com.gwtt.simulator.netconf.framing.ChunkFramingMechanism;
import com.gwtt.simulator.netconf.model.rpc.reply.RpcReply;
import com.gwtt.simulator.netconf.model.rpc.reply.RpcReplyOk;
import com.gwtt.simulator.netconf.model.rpc.request.Rpc;
import com.gwtt.simulator.netconf.util.XStreamUtil;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class NetconfMessageBus {

    private static final BlockingQueue<NetconfMessage> readQueue = new ArrayBlockingQueue<>(10000);
    private static final BlockingQueue<NetconfMessage> writeQueue = new ArrayBlockingQueue<>(10000);

    public static void addMessage(NetconfMessage message) {
        readQueue.add(message);
    }

    static {
        multiThreadExecuteTask(NetconfMessageBus::processMessage);
        multiThreadExecuteTask(NetconfMessageBus::writeWorkingInBackground);
    }

    private static void multiThreadExecuteTask(Runnable task) {
        ExecutorService es = Executors.newFixedThreadPool(8);
        for (int i = 0; i < 8; i++) {
            es.submit(task);
        }
    }

    public static void processMessage() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                NetconfMessage message = readQueue.take();
                if (message.getMessage().contains("<hello")) {
                    if (message.getMessage().contains("urn:ietf:params:netconf:base:1.1")) {
                        message.getClient().setEncoder(new ChunkFramingMechanism());
                    }
                    log.info("{} receive client hello message", message.getClient().getId());
                } else {
                    Rpc rpc = XStreamUtil.fromXML(message.getMessage(), Rpc.class);
                    RpcReply rpcReply = new RpcReply();
                    rpcReply.setMessageId(rpc.getMessageId());
                    rpcReply.setOk(new RpcReplyOk());
                    writeQueue.add(new NetconfMessage(message.getClient(), XStreamUtil.toXML(rpcReply)));
                }
            }
        } catch (Exception e) {
            log.error("failed to process message", e);
        }
    }

    public static void writeWorkingInBackground() {
        try {
            while (!Thread.currentThread().isInterrupted()) {
                NetconfMessage message = writeQueue.take();
                boolean write = message.getClient().write(message.getMessage());
                if (!write) {
                    writeQueue.put(message);
                }
            }
        } catch (Exception e) {
            log.error("failed to process message", e);
        }
    }

}
